package com.atguigu.edu.realtime.app.dwd;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetail { //下单表  dwd_trade_order_detail *********
    public static void main(String[] args) throws Exception {
        //0.环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //  1  获得topic_db
        String createTopicDbSQL="     CREATE TABLE topic_db (\n" +
                "        `database` STRING,\n" +
                "       `table` STRING,\n" +
                "        `type` STRING,\n" +
                "         `data` MAP<STRING,STRING>,\n" +
                "         `old` MAP<STRING,STRING> ,\n" +
                "          `proc_time` as proctime(), \n"+
                "         ts   STRING \n"+
                "          ) WITH (\n" +
                "          'connector' = 'kafka',\n" +
                "          'topic' = 'topic_db',\n" +
                "          'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "          'properties.group.id' = 'dwd_trade_order_detail',\n" +
                "          'scan.startup.mode' = 'group-offsets',\n" +
                "           'format' = 'json'\n" +
                "         )";
        tableEnv.executeSql(createTopicDbSQL);

        //  2  创建动态表
        //2.1   订单主表      order_info
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['province_id'] province_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_info'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_info", orderInfo);

        //2.2   订单明细表  order_detail
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['order_id'] order_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['origin_amount'] origin_amount,\n" +
                "data['coupon_reduce'] coupon_reduce,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['create_time'] create_time,\n" +
                "data['update_time'] update_time,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db` where `table` = 'order_detail' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);
/*        //2.3   购物券  order_detail_coupon
        Table orderDetailCoupon = tableEnv.sqlQuery("select\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['coupon_id'] coupon_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail_coupon'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);

        //2.4   活动    order_detail_activity
        Table orderDetailActivity = tableEnv.sqlQuery("select \n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['activity_id'] activity_id,\n" +
                "data['activity_rule_id'] activity_rule_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail_activity'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail_activity", orderDetailActivity);
        // 3   字典维度表
        String baseDicDDL = MySqlUtil.getBaseDicDDL();
        tableEnv.executeSql(baseDicDDL);*/
        //4   进行join操作
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "od.create_time,\n" +
                "od.update_time,\n" +
                "od.ts \n" +
                "from order_detail od \n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id"); // \n" +
        tableEnv.createTemporaryView("result_table", resultTable);
        //tableEnv.executeSql("select * from "+resultTable).print();
        // 5  目标表的定义
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "create_time string,\n" +
                "update_time string,\n" +
                "ts string,\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));
        // 6   写入目标表
        tableEnv.executeSql("" +
                "insert into dwd_trade_order_detail \n" +
                "select * from result_table");
    }
}


















